package com.zzf.concurrent.example.thread.pool;

import com.zzf.concurrent.example.thread.pool.exception.RejectException;
import com.zzf.concurrent.example.thread.pool.strategy.RejectStrategy;
import com.zzf.concurrent.example.util.TimeUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

/**
 * @author GaoFeng2017
 * @date 2019/8/9 9:42
 */
@Slf4j
public class ThreadPoolV1 extends Thread {

    /** 默认线程组 */
    private ThreadGroup threadPoolGroup = new ThreadGroup("THREAD-POOL");

    /** 线程名前缀 */
    private final static String THREAD_NAME_PREFIX = "THREAD-POOL-";

    /** 默认任务对列最大容量 */
    private final static int DEFAULT_MAX_TASK_SIZE = 40;

    /** 任务最大容量 */
    private int maxTaskSize;

    /** 用于定义线程名的自增序号 */
    private long threadSeqNumber = 0;

    /** 任务队列 */
    private final LinkedList<Runnable> workTask = new LinkedList<>();

    /** 线程容器 */
    private List<WorkThread> workThreads;

    /** 拒绝策略 */
    private RejectStrategy rejectStrategy = () -> {
        throw new RejectException("task queue size is exceeds limit!");
    };

    /** 线程池是否摧毁 */
    private volatile boolean isDestroy;

    /** 线程池最小数量 */
    private int minSize;

    /** 线程池活跃数量 */
    private int activeSize;

    /** 线程池最大数量 */
    private int maxSize;

    public ThreadPoolV1() {
        this(DEFAULT_MAX_TASK_SIZE, 4, 8, 12);
    }

    public ThreadPoolV1(int maxTaskSize, int minSize, int activeSize, int maxSize) {
        init(maxTaskSize, minSize, activeSize, maxSize);
    }

    public ThreadPoolV1(int maxTaskSize, int minSize, int activeSize, int maxSize , RejectStrategy rejectStrategy) {
        init(maxTaskSize, minSize, activeSize, maxSize);
        this.rejectStrategy = rejectStrategy;
    }

    private void init(int maxTaskSize, int minSize, int activeSize, int maxSize) {
        this.maxTaskSize = maxTaskSize;
        this.minSize = minSize;
        this.activeSize = activeSize;
        this.maxSize = maxSize;
        this.isDestroy = false;
        this.workThreads = new ArrayList<>(this.minSize + 1);
        for (int i = 0; i < this.minSize; i++) {
            workThreads.add(this.createWorkThread());
        }
    }

    private enum ThreadState {
        /** 线程状态-空闲 */
        FREE,
        /** 线程状态-执行中 */
        RUNNING,
        /** 线程状态-阻塞/等待 */
        BLOCKED,
        /** 线程状态-已终止 */
        DEAD
    }

    /** 添加任务 */
    public void addTask(Runnable task) {

        if (this.isDestroy) {
            throw new IllegalStateException("线程池已关闭!");
        }

        synchronized (workTask) {
            if (workTask.size() >= this.maxTaskSize) {
                rejectStrategy.reject();
            }
            this.workTask.add(task);
            workTask.notifyAll();
        }
    }

    @Override
    public void run() {
        while (!this.isDestroy) {
            int currentThreadSize = this.workThreads.size();
            int taskQueueSize = this.workTask.size();
            log.info("currentThreadSize: Min:{}, Max:{}, Active:{}, Current:{}, taskQueueSize: {}",
                    minSize, maxSize, activeSize, currentThreadSize, taskQueueSize);
            if (currentThreadSize < activeSize && taskQueueSize > currentThreadSize) {
                // 任务数量大于最大线程数量，创建新的工作线程
                for (int i = currentThreadSize; i < Math.min(taskQueueSize, activeSize); i++) {
                    this.workThreads.add(this.createWorkThread());
                }
            } else if (currentThreadSize < maxSize && taskQueueSize > currentThreadSize) {
                for (int i = currentThreadSize; i < Math.min(taskQueueSize, maxTaskSize); i++) {
                    this.workThreads.add(this.createWorkThread());
                }
            } else if (currentThreadSize > activeSize && taskQueueSize < currentThreadSize) {
                // 任务数量大于活跃线程数量，销毁工作线程
                synchronized (workThreads) {
                    int destroyCount = 0;
                    int diffCount = Math.min(currentThreadSize - taskQueueSize, maxSize - minSize);
                    Iterator<WorkThread> workThreadIterator = this.workThreads.iterator();
                    log.info("diffCount is {}", diffCount);

                    while(workThreadIterator.hasNext() && destroyCount < diffCount) {

                        WorkThread workThread = workThreadIterator.next();
                        if (workThread.getThreadState() == ThreadState.BLOCKED) {
                            workThread.interrupt();
                            workThread.close();
                            destroyCount++;
                            workThreadIterator.remove();
                            log.info("destroy thread {}, {}", workThread.getName(), destroyCount);
                        }
                    }

                }
            }

            TimeUtil.sleep(5000);

        }
    }

    private WorkThread createWorkThread() {
        WorkThread workThread = new WorkThread(threadPoolGroup, THREAD_NAME_PREFIX + (++threadSeqNumber));
        workThread.start();
        return workThread;
    }

    private class WorkThread extends Thread {

        /** 当前线程状态 */
        private volatile ThreadState state;

        public WorkThread(ThreadGroup group, String name) {
            super(group, name);
            this.state = ThreadState.FREE;
        }

        @Override
        public void run() {
            while (this.state != ThreadState.DEAD) {
                Runnable task = null;

                synchronized (workTask) {
                    while (workTask.isEmpty()) {
                        // 任务队列为空，等待
                        try {
                            this.state = ThreadState.BLOCKED;
                            workTask.wait();
                        } catch (InterruptedException e) {
                            log.info("{} is interrupted", this.getName());
                            return;
                        }
                    }

                    task = workTask.removeFirst();
                }

                // 接收到任务
                this.state = ThreadState.RUNNING;
                if (task != null) {
                    log.info("{} start execute task...", this.getName());
                    task.run();
                    log.info("{} finished execute.", this.getName());
                }

                this.state = ThreadState.FREE;
            }
        }

        public void close() {
            this.state = ThreadState.DEAD;
        }

        public ThreadState getThreadState() {
            return state;
        }
    }

    public void shutdown() {

        log.info("thread pool will be shutdown.");

        // 判断任务队列是否为空
        while (!workTask.isEmpty()) {
            TimeUtil.sleep(100);
            log.info("workTask size is {}", workTask.size());
        }

        // 遍历所有线程
        int closeCount = 0;

        while (closeCount < workThreads.size()) {
            synchronized (workThreads) {
                for (WorkThread workThread : workThreads) {
                    if (workThread.getThreadState() == ThreadState.BLOCKED) {
                        workThread.interrupt();
                        workThread.close();
                        closeCount++;
                        log.info("close thread {}, {}", workThread.getName(), closeCount);
                    } else {
                        TimeUtil.sleep(10);
                        // log.info("continue scan thread of closeable");
                    }
                }
            }
        }

        this.isDestroy = true;
        log.info("thread pool active count is {}", threadPoolGroup.activeCount());
        log.info("thread pool is shutdown.");
    }

    public int getPoolSize() {
        return this.workThreads.size();
    }

    public int getMaxTaskSize() {
        return maxTaskSize;
    }

    public boolean isDestroy() {
        return isDestroy;
    }
}
